package com.zhuchang.self_rpc.test.config;

import com.zhuchang.self_rpc.test.client.RpcClientHandler;
import com.zhuchang.self_rpc.test.client.RpcClientInitializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author RenPu
 * @Date 2021/8/17 17:04
 * @Version 1.0
 * @Description: 基于netty自定义RPC框架：连接管理器的编写
 **/
@Slf4j
public class RpcConnectMannager {


    //单例模式的饥饿加载方式
    private static volatile RpcConnectMannager RPC_CONNECT_MANNAGER = new RpcConnectMannager();

    //创建缓存连接对象作为key,具体业务逻辑处理类为value的容器
    private Map<InetSocketAddress, RpcClientHandler> connectHandlerMap = new ConcurrentHashMap<InetSocketAddress, RpcClientHandler>();

    //创建异步线程池，用于提交异步任务处理
    private ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(65536));

    //连接地址，业务处理的列表
    private CopyOnWriteArrayList<RpcClientHandler> copyOnWriteArrayList = new CopyOnWriteArrayList<RpcClientHandler>();

    //创建可重入锁
    private ReentrantLock reentrantLock = new ReentrantLock();

    //创建可重入锁的条件器
    private Condition condition = reentrantLock.newCondition();

    //设置锁的等待时间 单位毫秒
    private long connectTimeoutMills = 6000;

    //业务线程状态
    private volatile boolean isRunning = true;

    //创建原子对象，值设置为；0
    private volatile AtomicInteger indexId = new AtomicInteger(0);

    //默认值为cpu核数的2倍
    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(4);

    //创建私有构造方法
    public RpcConnectMannager() {
    }

    //创建对象的方法
    public static RpcConnectMannager getInstance() {
        return RPC_CONNECT_MANNAGER;
    }

    /**
     * 1：创建连接
     * 2：使用线程池采用异步连接
     * 3：从bootstrap.connect 创建真正的连接
     * 4：创建听听器，监听连接的状态
     * 5：连接失败以及连接成功后的处理逻辑
     * 6：对于连接做一个缓存管理，避免资源的额浪费
     */


    /**
     * 发起连接的方法
     * 参数使用final修饰，防止参数的篡改：serverAddress=ip+port
     *
     * @param serverAddress //1. 异步连接 线程池 真正的发起连接，连接失败监听，连接成功监听
     *                      //2. 对于连接进来的资源做一个缓存（做一个管理）updateConnectedServer
     *                      /**
     *                      $connect 发起连接方法
     * @param serverAddress
     */
    public void connect(final String serverAddress) {
        List<String> allServerAddress = Arrays.asList(serverAddress.split(","));
        updateConnectServer(allServerAddress);
    }

    /**
     * add connect List<String> serverAddress
     *
     * @param serverAddress
     */
    public void connect(List<String> serverAddress) {
        updateConnectServer(serverAddress);
    }


    /**
     * 对于连接做一个缓存管理的方法
     * 例如：192.168.11.111:8848,192.168.11.166:9697
     *
     * @param allServerAddress
     */
    public void updateConnectServer(List<String> allServerAddress) {

        //创建连接对象的容器为HashSet，防止重复连接耗费资源
        HashSet<InetSocketAddress> inetSocketAddresses = new HashSet<InetSocketAddress>();

        if (CollectionUtils.isNotEmpty(allServerAddress)) {
            for (int i = 0; i < allServerAddress.size(); i++) {
                String[] array = allServerAddress.get(i).split(":");
                if (array.length == 2) {
                    String host = array[0];
                    int port = Integer.parseInt(array[1]);

                    //java.net下的对象，创建连接对象
                    InetSocketAddress inetSocketAddress = new InetSocketAddress(host, port);
                    inetSocketAddresses.add(inetSocketAddress);
                }
            }


            //遍历发起连接
            for (InetSocketAddress inetSocketAddress : inetSocketAddresses) {
                if (!connectHandlerMap.keySet().contains(inetSocketAddress)) {
                    //发起异步连接
                    connectAsyc(inetSocketAddress);
                }
            }

            //TODO 如果allServerAddress容器里，不包含的连接，就要从缓存容器里移除
            for (int i = 0; i < copyOnWriteArrayList.size(); i++) {
                RpcClientHandler rpcClientHandler = copyOnWriteArrayList.get(i);
                InetSocketAddress remotePeer = (InetSocketAddress) rpcClientHandler.getRemotePeer();

                if (!inetSocketAddresses.contains(remotePeer)) {
                    log.info(" remove invalid server node " + remotePeer);
                    RpcClientHandler handler = connectHandlerMap.get(remotePeer);
                    if (handler != null) {
                        handler.close();
                        connectHandlerMap.remove(remotePeer);
                    }
                    copyOnWriteArrayList.remove(rpcClientHandler);

                }
            }

        } else {

            //添加告警
            log.error("no available server address ! ! !");

            // 清楚所有缓存信息
            clearConnected();
        }
    }


    /**
     * 异步连接的方法
     *
     * @param remotePeer
     */
    private void connectAsyc(final InetSocketAddress remotePeer) {

        threadPoolExecutor.submit(new Runnable() {
            public void run() {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(eventLoopGroup)
                        .channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)     //此设置可以禁止数据包，自由合并的操作
                        .handler(new RpcClientInitializer());
                connect(bootstrap, remotePeer);
            }
        });

    }

    /**
     * 执行连接的方法
     *
     * @param bootstrap
     * @param remotePeer
     */
    private void connect(final Bootstrap bootstrap, final InetSocketAddress remotePeer) {

        //1：调用bootstrap.connect真正建立连接
        final ChannelFuture channelFuture = bootstrap.connect(remotePeer);

        //2：添加监听器监听连接失败，清除资源再次进行重新连接
        channelFuture.channel().closeFuture().addListener(new ChannelFutureListener() {

            public void operationComplete(ChannelFuture future) throws Exception {
                log.warn("connect fail !!!, to connect !");

                /** 释放资源并设置 3秒，进行重新 **/
                future.channel().eventLoop().schedule(new Runnable() {
                    public void run() {
                        clearConnected();
                        connect(bootstrap, remotePeer);
                    }
                }, 3, TimeUnit.SECONDS);
            }
        });

        //3：连接成功，将连接对象放入缓存
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    log.info("successfully connect to remote server, remote peer:" + remotePeer);
                    RpcClientHandler rpcClientHandler = future.channel().pipeline().get(RpcClientHandler.class);
                    addHandler(rpcClientHandler);
                }

            }
        });


    }


    /**
     * 将成功连接的，业务处理器，存在指定的容器里
     * copyOnWriteArrayList & connectHandlerMap
     *
     * @param rpcClientHandler
     */
    private void addHandler(RpcClientHandler rpcClientHandler) {
        copyOnWriteArrayList.add(rpcClientHandler);
        InetSocketAddress inetSocketAddress = (InetSocketAddress) rpcClientHandler.getRemotePeer();
        //connectHandlerMap.put(inetSocketAddress,rpcClientHandler);

        //sign Alavilable Handler 唤醒可用的业务执行器
        signalAlavilableHandler();

    }


    /**
     * 释放连接失败时，及时释放资源，清空缓冲
     * 1:释放缓存connectHandlerMap的资源
     * 2：释放业务处理列表容器的资源copyOnWriteArrayList
     **/
    private void clearConnected() {

        //遍历失败连接的任务处理器列表，获取对应的连接对象，然后根据连接对象，从缓存connectMap里进行移除
        for (RpcClientHandler rpcClientHandler : copyOnWriteArrayList) {
            SocketAddress remotePeer = rpcClientHandler.getRemotePeer();
            RpcClientHandler handler = connectHandlerMap.get(remotePeer);
            if (handler != null) {
                //先关闭连接，在从缓存释放资源
                handler.close();
                connectHandlerMap.remove(remotePeer);
            }
        }

        //业务处理列表容器的资源copyOnWriteArrayList
        copyOnWriteArrayList.clear();
    }


    /**
     * sign Alavilable Handler 唤醒可用的业务执行器
     */
    public void signalAlavilableHandler() {

        reentrantLock.lock();


        try {

            //唤醒对方
            condition.signalAll();
        } finally {
            reentrantLock.unlock();
        }


    }


    /**
     * 等待新连接通知的方法
     *
     * @return
     * @throws InterruptedException
     */
    private boolean waitAlavilableHandler() throws InterruptedException {

        reentrantLock.lock();

        try {
            //唤醒对方
            return condition.await(this.connectTimeoutMills, TimeUnit.MICROSECONDS);
        } finally {
            reentrantLock.unlock();
        }


    }


    /**
     * 选择一个实际的业务进行处理
     *
     * @return
     */
    public RpcClientHandler chooseHandler() {

        //1：防止线程安全问题复制出 copyOnWriteArrayList
        CopyOnWriteArrayList<RpcClientHandler> handlers = (CopyOnWriteArrayList<RpcClientHandler>) copyOnWriteArrayList.clone();

        int size = handlers.size();

        //2: while条件判断，根据runing和copyOnWriteArrayList.size()
        while (isRunning && handlers.size() <= 0) {
            //3:调用waitAlavilableHandler方法，查看是否有新连接进入，3.1:有新连接进入将copyOnWriteArrayList.size()和copyOnWriteArrayList重新赋值
            try {
                boolean alavilableHandler = waitAlavilableHandler();
                if (alavilableHandler) {
                    handlers = (CopyOnWriteArrayList<RpcClientHandler>) copyOnWriteArrayList.clone();
                    size = handlers.size();
                }
            } catch (InterruptedException e) {
                log.error("waiting for available node is interrupted !");
                throw new RuntimeException("no connect any server !");

            }

        }

        //防止算术by zero异常的抛出
        if (!isRunning) {
            return null;

        }

        //4：跳出while循环定义取模规律，从copyOnWriteArrayList取出新连接对象,进行实际的业务处理
        return handlers.get((indexId.getAndAdd(1) + size) % size);
    }


    /**
     * 停止方法
     */
    public void stop() {

        //设置运行状态为false
        isRunning = false;

        //遍历copyOnWriteArrayList 容器对业务处理器对象，进行释放资源
        for (int i = 0; i < copyOnWriteArrayList.size(); i++) {
            RpcClientHandler rpcClientHandler = copyOnWriteArrayList.get(i);
            rpcClientHandler.close();
        }

        //停止方法前，再次唤醒业务操作，防止算术by zero异常的抛出
        signalAlavilableHandler();

        //释放线程池资源以及eventLoopGroup事件组资源
        threadPoolExecutor.shutdown();
        eventLoopGroup.shutdownGracefully();
    }


    /**
     * 发起重连方法
     *
     * @param handler
     * @param remotePeer
     */
    public void reconnect(final RpcClientHandler handler, final SocketAddress remotePeer) {

        //1:非空判断
        if (handler != null) {

            //2：从copyOnWriteArrayList和connectHandlerMap容器里移除资源
            copyOnWriteArrayList.remove(handler);
            SocketAddress remotePeer1 = handler.getRemotePeer();
            connectHandlerMap.remove(remotePeer1);
        }

        //3：调用重连方法
        connectAsyc((InetSocketAddress) remotePeer);

    }

}
